﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Reflection;
using System.Text;
using ESoft.Extensions;
using ESoft.Helpers;
using Newtonsoft.Json.Linq;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;

namespace ESoft.RabbitMQ
{
    /// <summary>
    /// RabbitMQ服务
    /// </summary>
    public class RabbitMQService : IRabbitMQService
    {
        private const string Exchange = "topic";
        private readonly IChannelObjectPool _channelObjectPool;

        /// <summary>
        /// 初始化
        /// </summary>
        /// <param name="channelObjectPool"></param>
        public RabbitMQService(IChannelObjectPool channelObjectPool)
        {
            _channelObjectPool = channelObjectPool;
        }

        /// <summary>
        /// 启动消费者
        /// </summary>
        /// <param name="consumers"></param>
        /// <param name="assemblies"></param>
        public void StartConsumers(IEnumerable<IConsumer> consumers, params Assembly[] assemblies)
        {
            consumers.Each(bizConsumer =>
            {
                var rabbitAttris = bizConsumer.GetType().GetCustomAttributes<RabbitBindingAttribute>();
                rabbitAttris.Each(rabbit =>
                {
                    var channel = _channelObjectPool.Get();
                    try
                    {
                        channel.ExchangeDeclare(Exchange, ExchangeType.Topic);
                        channel.QueueDeclare(rabbit.QueueName, false, false, false, null);
                        channel.QueueBind(rabbit.QueueName, rabbit.Exchange, rabbit.RouteKey, null);
                        var eventingBasicConsumer = new EventingBasicConsumer(channel);
                        eventingBasicConsumer.Received += (ch, ea) =>
                        {
                            var pro = ea.BasicProperties;
                            if (pro.Headers.ContainsKey("className"))
                            {
                                var className = Encoding.UTF8.GetString((byte[])pro.Headers["className"]);
                                var body = ea.Body.ToArray();

                                var jObject = ByteConvertHelper.Bytes2Object<JObject>(body);
                                var types = assemblies.SelectMany(assembly => assembly.GetTypes()).Where(type => type.FullName.Equals(className)).ToArray();

                                if (types.Length == 1)
                                {
                                    IPayload payload = jObject.ToObject(types[0]) as IPayload;
                                    bizConsumer.OnRecieve(payload);
                                }
                            }

                            channel.BasicAck(ea.DeliveryTag, false);
                        };

                        channel.BasicConsume(rabbit.QueueName, false, eventingBasicConsumer);
                    }
                    catch (Exception)
                    {
                        _channelObjectPool.Return(channel);
                        throw;
                    }
                });
            });
        }

        /// <summary>
        /// 发送消息
        /// </summary>
        /// <param name="payload"></param>
        /// <param name="routeKey"></param>
        public void Send(IPayload payload, string routeKey)
        {
            var channel = _channelObjectPool.Get();
            try
            {
                IBasicProperties props = CreateProperties(payload, channel);
                byte[] payloadBytes = ByteConvertHelper.Object2Bytes(payload);
                channel.ExchangeDeclare(Exchange, ExchangeType.Topic);
                channel.BasicPublish(Exchange, routeKey, props, payloadBytes);
            }
            finally
            {
                _channelObjectPool.Return(channel);
            }
        }

        private static IBasicProperties CreateProperties(IPayload payload, IModel channel)
        {
            IBasicProperties props = channel.CreateBasicProperties();
            props.ContentType = "application/json";
            props.DeliveryMode = 2;
            props.CorrelationId = StringHelper.GetId();
            props.Headers = new Dictionary<string, object> { { "className", payload.GetType().FullName } };
            return props;
        }

        /// <summary>
        /// 发送延时消息
        /// </summary>
        /// <param name="payload"></param>
        /// <param name="routeKey"></param>
        /// <param name="delayMs"></param>
        public void SendDelay(IPayload payload, string routeKey, int delayMs)
        {
            var channel = _channelObjectPool.Get();
            try
            {
                string delayRoute = "delay." + routeKey + "_" + delayMs;
                string delayExchange = "delay." + Exchange;
                channel.ExchangeDeclare(Exchange, ExchangeType.Topic);

                // 定义死信
                IDictionary<string, object> arguments = new Dictionary<string, object>
            {
                { "x-dead-letter-exchange", Exchange },
                { "x-dead-letter-routing-key", routeKey }
            };

                channel.ExchangeDeclare(delayExchange, ExchangeType.Topic, true, false, null);
                channel.QueueDeclare(delayRoute, true, false, false, arguments);
                channel.QueueBind(delayRoute, delayExchange, delayRoute);

                IBasicProperties props = CreateProperties(payload, channel);
                props.Expiration = delayMs.ToString();
                byte[] payloadBytes = ByteConvertHelper.Object2Bytes(payload);
                channel.BasicPublish(delayExchange, delayRoute, props, payloadBytes);
            }
            finally
            {
                _channelObjectPool.Return(channel);
            }
        }
    }
}
